package org.zjt.flink;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.configuration.Configuration;

import java.io.File;

/**
 * Description:
 *
 * @author juntao.zhang
 * Date: 2018-10-09 下午5:59
 * @see
 */
public class hdfsFileCache {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

// register a file from HDFS
        env.registerCachedFile("hdfs:///path/to/your/file", "hdfsFile");

// register a local executable file (script, executable, ...)
        env.registerCachedFile("file:///path/to/exec/file", "localExecFile", true);

        env.fromElements(3482,423,432,45,4,543,543,53).map(new MyMapper());

        env.execute();
    }


     final static class MyMapper extends RichMapFunction<Integer, Integer> {

        @Override
        public void open(Configuration config) {

            // access cached file via RuntimeContext and DistributedCache
            File myFile = getRuntimeContext().getDistributedCache().getFile("hdfsFile");
            // read the file (or navigate the directory)

        }

        @Override
        public Integer map(Integer value) throws Exception {
            // use content of cached file
            return value + 10 ;
        }
    }
}
